package com.atguigu.market_analysis

import java.util.{Random, UUID}

import org.apache.flink.streaming.api.functions.source.{RichSourceFunction, SourceFunction}

/**
 * 自定义source源
 *
 * Project: UserBehaviorAnalysis
 * Package: com.atguigu.market_analysis
 * Version: 1.0
 *
 * Created by  WangJX  on 2019/12/12 15:59
 */
class MySource extends RichSourceFunction[MarketingUserBehavior]{

  private var running: Boolean = true

  private val rand: Random = new Random()

  private val channelSet: Seq[String] = Seq("AppStore", "XiaomiStore", "HuaweiStore", "weibo", "wechat", "tieba")
  private val behaviorTypes: Seq[String] = Seq("BROWSE", "CLICK", "PURCHASE", "UNINSTALL")

  override def run(ctx: SourceFunction.SourceContext[MarketingUserBehavior]): Unit = {

    while (running) {
      val id = UUID.randomUUID().toString
      val behavior = behaviorTypes(rand.nextInt(behaviorTypes.size))
      val channel = channelSet(rand.nextInt(channelSet.size))
      val timestamp = System.currentTimeMillis()

      Thread.sleep(100L)

      ctx.collect(MarketingUserBehavior(id, behavior, channel, timestamp))
    }

  }

  override def cancel(): Unit = running = false
}


/**
 *
 * @param userId  用户id
 * @param behavior  行为
 * @param channel 渠道
 * @param timestamp 时间
 */
case class MarketingUserBehavior(
                                  userId: String,
                                  behavior: String,
                                  channel: String,
                                  timestamp: Long
                                ){

}